{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Tce3stUlHN0L"
      },
      "source": [
        "##### Copyright 2019 The TensorFlow Authors.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "tuOe1ymfHZPu"
      },
      "outputs": [],
      "source": [
        "#@title Licensed under the Apache License, Version 2.0 (the \"License\");\n",
        "# you may not use this file except in compliance with the License.\n",
        "# You may obtain a copy of the License at\n",
        "#\n",
        "# https://www.apache.org/licenses/LICENSE-2.0\n",
        "#\n",
        "# Unless required by applicable law or agreed to in writing, software\n",
        "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
        "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
        "# See the License for the specific language governing permissions and\n",
        "# limitations under the License."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "MfBg1C5NB3X0"
      },
      "source": [
        "# Обучение нескольких рабочих процессов с помощью Keras\n",
        "\n",
        "<table class=\"tfo-notebook-buttons\" align=\"left\">\n",
        "  <td>\n",
        "    <a target=\"_blank\" href=\"https://www.tensorflow.org/tutorials/distribute/multi_worker_with_keras\"><img src=\"https://www.tensorflow.org/images/tf_logo_32px.png\" />Смотрите на TensorFlow.org</a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a target=\"_blank\" href=\"https://colab.research.google.com/github/tensorflow/docs-l10n/blob/master/site/ru/tutorials/distribute/multi_worker_with_keras.ipynb\"><img src=\"https://www.tensorflow.org/images/colab_logo_32px.png\" />Запустите в Google Colab</a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a target=\"_blank\" href=\"https://github.com/tensorflow/docs-l10n/blob/master/site/ru/tutorials/distribute/multi_worker_with_keras.ipynb\"><img src=\"https://www.tensorflow.org/images/GitHub-Mark-32px.png\" />Изучайте код на GitHub</a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a href=\"https://storage.googleapis.com/tensorflow_docs/docs-l10n/site/ru/tutorials/distribute/multi_worker_with_keras.ipynb\"><img src=\"https://www.tensorflow.org/images/download_logo_32px.png\" />Скачайте ноутбук</a>\n",
        "  </td>\n",
        "</table>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "d3874b352184"
      },
      "source": [
        "Note: Вся информация в этом разделе переведена с помощью русскоговорящего Tensorflow сообщества на общественных началах. Поскольку этот перевод не является официальным, мы не гарантируем что он на 100% аккуратен и соответствует [официальной документации на английском языке](https://www.tensorflow.org/?hl=en). Если у вас есть предложение как исправить этот перевод, мы будем очень рады увидеть pull request в [tensorflow/docs](https://github.com/tensorflow/docs) репозиторий GitHub. Если вы хотите помочь сделать документацию по Tensorflow лучше (сделать сам перевод или проверить перевод подготовленный кем-то другим), напишите нам на [docs-ru@tensorflow.org list](https://groups.google.com/a/tensorflow.org/forum/#!forum/docs-ru)."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xHxb-dlhMIzW"
      },
      "source": [
        "## Введение\n",
        "\n",
        "В этом руководстве демонстрируется распределенное обучение нескольких воркеров с помощью модели Keras с использованием `tf.distribute.Strategy`, в частности `tf.distribute.experimental.MultiWorkerMirroredStrategy`. С помощью этой стратегии модель Keras, которая была разработана для работы с одним рабочим процессом, может беспрепятственно работать с несколькими воркерами с минимальным изменением кода.\n",
        "\n",
        "[Распределенное обучение в TensorFlow](../../guide/distribute_training.ipynb) - руководство по распределенным стратегиям, поддерживаемым TensorFlow, для тех, кто заинтересован в более глубоком понимании `tf.distribute.Strategy`.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "MUXex9ctTuDB"
      },
      "source": [
        "## Установка и импорт"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "bnYxvfLD-LW-"
      },
      "outputs": [],
      "source": [
        "import json\n",
        "import os\n",
        "import sys"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Zz0EY91y3mxy"
      },
      "source": [
        "Перед импортом TensorFlow внесите несколько изменений в рабочее окружение.\n",
        "\n",
        "Отключите все графические процессоры. Это предотвращает ошибки, вызванные тем, что все воркеры пытаются использовать один и тот же графический процессор. В реальном приложении каждый воркер будет работать на разных машинах."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "685pbYEY3jGC"
      },
      "outputs": [],
      "source": [
        "os.environ[\"CUDA_VISIBLE_DEVICES\"] = \"-1\""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7X1MS6385BWi"
      },
      "source": [
        "Сбросьте переменную окружения TF_CONFIG, о ней мы поговорим позже."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "WEJLYa2_7OZF"
      },
      "outputs": [],
      "source": [
        "os.environ.pop('TF_CONFIG', None)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Rd4L9Ii77SS8"
      },
      "source": [
        "Убедитесь, что текущий каталог находится в путях Python. Это позже позволит ноутбуку импортировать файлы, записанные с помощью `%%writefile`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "hPBuZUNSZmrQ"
      },
      "outputs": [],
      "source": [
        "if '.' not in sys.path:\n",
        "  sys.path.insert(0, '.')"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "pDhHuMjb7bfU"
      },
      "source": [
        "Теперь импортируемt TensorFlow."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vHNvttzV43sA"
      },
      "outputs": [],
      "source": [
        "import tensorflow as tf"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0S2jpf6Sx50i"
      },
      "source": [
        "### Датасет и определение модели"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fLW6D2TzvC-4"
      },
      "source": [
        "Next create an `mnist.py` file with a simple model and dataset setup. This python file will be used by the worker-processes in this tutorial:\n",
        "\n",
        "Теперь создайте файл `mnist.py` с простой моделью и установками для датасета. Этот файл будет использоваться рабочими процессами в этом руководстве:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "dma_wUAxZqo2"
      },
      "outputs": [],
      "source": [
        "%%writefile mnist.py\n",
        "\n",
        "import os\n",
        "import tensorflow as tf\n",
        "import numpy as np\n",
        "\n",
        "def mnist_dataset(batch_size):\n",
        "  (x_train, y_train), _ = tf.keras.datasets.mnist.load_data()\n",
        "  # `x` это список типа uint8 и содержит числа в диапазоне [0, 255].\n",
        "  # вам нужно конвертировать их во float32 и значениями в диапазоне [0, 1]\n",
        "  x_train = x_train / np.float32(255)\n",
        "  y_train = y_train.astype(np.int64)\n",
        "  train_dataset = tf.data.Dataset.from_tensor_slices(\n",
        "      (x_train, y_train)).shuffle(60000).repeat().batch(batch_size)\n",
        "  return train_dataset\n",
        "\n",
        "def build_and_compile_cnn_model():\n",
        "  model = tf.keras.Sequential([\n",
        "      tf.keras.Input(shape=(28, 28)),\n",
        "      tf.keras.layers.Reshape(target_shape=(28, 28, 1)),\n",
        "      tf.keras.layers.Conv2D(32, 3, activation='relu'),\n",
        "      tf.keras.layers.Flatten(),\n",
        "      tf.keras.layers.Dense(128, activation='relu'),\n",
        "      tf.keras.layers.Dense(10)\n",
        "  ])\n",
        "  model.compile(\n",
        "      loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),\n",
        "      optimizer=tf.keras.optimizers.SGD(learning_rate=0.001),\n",
        "      metrics=['accuracy'])\n",
        "  return model"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2UL3kisMO90X"
      },
      "source": [
        "Попробуйте обучить модель на небольшом количестве эпох и понаблюдайте за результатами одного воркера, чтобы убедиться, что все работает правильно. По мере обучения потери должны уменьшаться, а точность - увеличиваться."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6Qe6iAf5O8iJ"
      },
      "outputs": [],
      "source": [
        "import mnist\n",
        "\n",
        "batch_size = 64\n",
        "single_worker_dataset = mnist.mnist_dataset(batch_size)\n",
        "single_worker_model = mnist.build_and_compile_cnn_model()\n",
        "single_worker_model.fit(single_worker_dataset, epochs=3, steps_per_epoch=70)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "JmgZwwymxqt5"
      },
      "source": [
        "## Многопользовательская конфигурация\n",
        "\n",
        "Теперь давайте познакомимся с обучением нескольких воркеров. В TensorFlow переменная среды окружения `TF_CONFIG` требуется для обучения на нескольких машинах, каждая из которых, может иметь свою роль. `TF_CONFIG` - это строка JSON, используемая для указания конфигурации кластера для каждого воркера, являющегося частью кластера.\n",
        "\n",
        "Вот пример конфигурации:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "XK1eTYvSZiX7"
      },
      "outputs": [],
      "source": [
        "tf_config = {\n",
        "    'cluster': {\n",
        "        'worker': ['localhost:12345', 'localhost:23456']\n",
        "    },\n",
        "    'task': {'type': 'worker', 'index': 0}\n",
        "}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "JjgwJbPKZkJL"
      },
      "source": [
        "Вот тот же `TF_CONFIG`, сериализованный как JSON-строка:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "yY-T0YDQZjbu"
      },
      "outputs": [],
      "source": [
        "json.dumps(tf_config)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "AUBmYRZqxthH"
      },
      "source": [
        "`TF_CONFIG` состоит из двух компонентов: `cluster` и `task`.\n",
        "\n",
        "* `cluster` одинаков для всех воркеров и содержит информацию об обучающем кластере вцелом, это словарь, состоящий из разных типов заданий, например `worker`. При обучении нескольких воркеров с помощью `MultiWorkerMirroredStrategy` обычно определяется один `worker`, который берет на себя немного больше ответственности, например, сохранение контрольной точки и написание сводного файла для TensorBoard в дополнение к тому, что делает обычный `worker`. Такого воркера называют `главным`, и обычно `worker` с индексом 0 назначается главным (именно так реализуется `tf.distribute.Strategy`).\n",
        "\n",
        "* `task` содержит информацию о текущей задаче и отличается для каждого воркера. Он определяет `type` и `index` этого воркера."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "8YFpxrcsZ2xG"
      },
      "source": [
        "В этом примере вы устанавливаете `type` задачи в значение` worker` и `index` задачи в значение `0`. Эта машина является первым воркером, который будет главным в кластере и будет выполнять больше работы, чем другие. Обратите внимание, что на других машинах также должна быть установлена переменная среды `TF_CONFIG`, и у нее должен быть такой же словарь с ключем `cluster`, но с другим типом задачи или индекса, в зависимости от роли этих машин."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "aogb74kHxynz"
      },
      "source": [
        "В качестве примера в этом руководстве показано, как можно установить `TF_CONFIG` с двумя рабочими процессами на `localhost`. На практике пользователи могут создать несколько рабочих процессов на внешних IP-адресах/портах и сконфигурировать `TF_CONFIG` соответственно для каждого воркера.\n",
        "\n",
        "В этом примере вы будете использовать 2 воркера, `TF_CONFIG` первого воркера показан выше. Для второго воркера вы должны установить `tf_config ['task'] ['index'] = 1`"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "f83FVYqDX3aX"
      },
      "source": [
        "В данном случае `tf_config` - это просто локальная переменная в Python. Чтобы действительно использовать его для настройки обучения, этот словарь нужно сериализовать в JSON и записать в переменную окружения `TF_CONFIG`."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cIlkfWmjz1PG"
      },
      "source": [
        "### Переменные окружения и подпроцессы в ноутбуках"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FcjAbuGY1ACJ"
      },
      "source": [
        "Подпроцессы наследуют переменные окружения от своего родителя. Таким образом, если вы установите переменную окружения `jupyter notebook` в этом процессе:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "PH2gHn2_0_U8"
      },
      "outputs": [],
      "source": [
        "os.environ['GREETINGS'] = 'Hello TensorFlow!'"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "gQkIX-cg18md"
      },
      "source": [
        "Вы можете получить доступ к этой переменной из подпроцессов:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "pquKO6IA18G5"
      },
      "outputs": [],
      "source": [
        "%%bash\n",
        "echo ${GREETINGS}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "af6BCA-Y2fpz"
      },
      "source": [
        "В следующем разделе вы будете использовать переменные окружения, чтобы передать `TF_CONFIG` рабочим подпроцессам. Вы никогда не запустите свои рабочие процессы таким образом в реальном окружении, но этого достаточно для данного руководства: продемонстрировать минимальный пример с несколькими воркерам."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "UhNtHfuxCGVy"
      },
      "source": [
        "## Выбор правильной стратегии\n",
        "\n",
        "В TensorFlow есть две основные формы распределенного обучения:\n",
        "\n",
        "* Синхронное обучение, при котором шаги обучения синхронизируются между воркерами и репликами, а также\n",
        "* Асинхронное обучение, при котором шаги обучения не синхронизированы.\n",
        "\n",
        "В этом руководстве будет показана стратегия `tf.distribute.experimental.MultiWorkerMirroredStrategy`, рекомендуемая для синхронного обучения нескольких воркеров.\n",
        "\n",
        "`MultiWorkerMirroredStrategy` создает копии всех переменных в слоях модели на каждом устройстве для всех рабочих процессов. Он использует `CollectiveOps`, операцию TensorFlow для коллективного общения, чтобы агрегировать градиенты и синхронизировать переменные. В [руководстве по `tf.distribute.Strategy`](../../guide/distributed_training.ipynb) есть более подробная информация."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "1uFSHCJXMrQ-"
      },
      "outputs": [],
      "source": [
        "strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "N0iv7SyyAohc"
      },
      "source": [
        "Примечание: сначала анализируется `TF_CONFIG`, затем, во время вызова `MultiWorkerMirroredStrategy()`, запускаются GRPC сервера TensorFlow. Поэтому переменная окружения `TF_CONFIG` должна быть установлена **до** создания экземпляра `tf.distribute.Strategy`. В нашем случае `TF_CONFIG` еще не установлен и вышеуказанная стратегия фактически представляет собой обучение одного воркера."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FMy2VM4Akzpr"
      },
      "source": [
        "`MultiWorkerMirroredStrategy` предоставляет несколько реализаций через [`CollectiveCommunication`](https://github.com/tensorflow/tensorflow/blob/a385a286a930601211d78530734368ccb415bee4/tensorflow/python/distribute/cross_device_ops.parameter). `RING` реализует группы на основе кольца, используя gRPC в качестве уровня связи между хостами. NCCL использует [NCCL Nvidia](https://developer.nvidia.com/nccl) для реализации групп. `AUTO` откладывает выбор на время выполнения. Наилучший выбор реализации групп зависит от количества и типа графических процессоров, а также от сетевого соединения в кластере."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "H47DDcOgfzm7"
      },
      "source": [
        "## Тренировка модели\n",
        "\n",
        "Интеграция API `tf.distribute.Strategy` в `tf.keras` требует единственное изменение для распределния обучения между несколькими воркерами, - это построение модели и вызов `model.compile()` внутри `strategy.scope ()`. Контекст(scope) распределнных стратегий определяет, как и где создаются переменные, а в случае `MultiWorkerMirroredStrategy` создаваемые переменные - это `MirroredVariable`, и они реплицируются на каждый из рабочих процессов."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "wo6b9wX65glL"
      },
      "outputs": [],
      "source": [
        "with strategy.scope():\n",
        "  # Создание и компиляция модели должны быть внутри `strategy.scope()`.\n",
        "  multi_worker_model = mnist.build_and_compile_cnn_model()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Mhq3fzyR5hTw"
      },
      "source": [
        "Примечание. В настоящее время существует ограничение в `MultiWorkerMirroredStrategy` - операторы TensorFlow должны быть созданы после создания экземпляра стратегии. Если вы видите ошибку `RuntimeError: Collective ops must be configured at program startup`, попробуйте создать экземпляр `MultiWorkerMirroredStrategy` в начале программы и поместите код, который создает операции, после создания стратегии."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jfYpmIxO6Jck"
      },
      "source": [
        "Чтобы по-настоящему запустить `MultiWorkerMirroredStrategy`, вам нужно запустить рабочие процессы и передать им `TF_CONFIG`.\n",
        "\n",
        "Подобно скрипту `mnist.py`, написанному ранее, вот скрипт `main.py`, который будет запускать каждый из воркеров:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "BcsuBYrpgnlS"
      },
      "outputs": [],
      "source": [
        "%%writefile main.py\n",
        "\n",
        "import os\n",
        "import json\n",
        "\n",
        "import tensorflow as tf\n",
        "import mnist\n",
        "\n",
        "per_worker_batch_size = 64\n",
        "tf_config = json.loads(os.environ['TF_CONFIG'])\n",
        "num_workers = len(tf_config['cluster']['worker'])\n",
        "\n",
        "strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()\n",
        "\n",
        "global_batch_size = per_worker_batch_size * num_workers\n",
        "multi_worker_dataset = mnist.mnist_dataset(global_batch_size)\n",
        "\n",
        "with strategy.scope():\n",
        "    multi_worker_model = mnist.build_and_compile_cnn_model()\n",
        "\n",
        "\n",
        "multi_worker_model.fit(multi_worker_dataset, epochs=3, steps_per_epoch=70)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Aom9xelvJQ_6"
      },
      "source": [
        "Обратите внимание, что в приведенном выше коде, для параметра `global_batch_size`, передаваемого в `Dataset.batch`, установлено значение `per_worker_batch_size * num_workers`. Это гарантирует, что каждый рабочий процесс обрабатывает партии размером `per_worker_batch_size`, независимо от количества рабочих."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lHLhOii67Saa"
      },
      "source": [
        "Текущий каталог теперь содержит оба скрипта:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "bi6x05Sr60O9"
      },
      "outputs": [],
      "source": [
        "%%bash\n",
        "ls *.py"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qmEEStPS6vR_"
      },
      "source": [
        "Теперь нужно сериализовать в json переменную `TF_CONFIG` и добавить ее в переменные окружения:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "9uu3g7vV7Bbt"
      },
      "outputs": [],
      "source": [
        "os.environ['TF_CONFIG'] = json.dumps(tf_config)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "MsY3dQLK7jdf"
      },
      "source": [
        "Теперь вы можете запустить рабочий процесс, который будет запускать `main.py` и использовать `TF_CONFIG`:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "txMXaq8d8N_S"
      },
      "outputs": [],
      "source": [
        "# предварительно завершив предыдущий запуск\n",
        "%killbgscripts"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "qnSma_Ck7r-r"
      },
      "outputs": [],
      "source": [
        "%%bash --bg\n",
        "python3 main.py &> job_0.log"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZChyazqS7v0P"
      },
      "source": [
        "Несколько уточнения по команде выше:\n",
        "\n",
        "1. Команда `%%bash` является [\"магией\" jupyter notebook](https://ipython.readthedocs.io/en/stable/interactive/magics.html), которая используется для выполнения некоторых команд bash.\n",
        "2. Флаг `--bg` нужен для запуска процесса `bash` в фоновом режиме, потому что этот рабочий процесс не завершается, а ждет всех воркеров перед запуском.\n",
        "\n",
        "Фоновый рабочий процесс не будет выводить логи в ноутбук, поэтому `&>` перенаправляет его вывод в файл, чтобы позже вы смогли просмотреть процесс работы.\n",
        "\n",
        "Подождите несколько секунд, пока процесс запустится:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Hm2yrULE9281"
      },
      "outputs": [],
      "source": [
        "import time\n",
        "time.sleep(10)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZFPoNxg_9_Mx"
      },
      "source": [
        "Теперь посмотрим, что было выведено в лог-файл воркера:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vZEOuVgQ9-hn"
      },
      "outputs": [],
      "source": [
        "%%bash\n",
        "cat job_0.log"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "RqZhVF7L_KOy"
      },
      "source": [
        "В последней строке лога должно быть написано: `Started server with target: grpc://localhost:12345`. Первый воркер готов и ждет, пока все остальные воркеры будут готовы запуститься."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Pi8vPNNA_l4a"
      },
      "source": [
        "Обновите `tf_config`, чтобы его подхватил второй рабочий процесс:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "lAiYkkPu_Jqd"
      },
      "outputs": [],
      "source": [
        "tf_config['task']['index'] = 1\n",
        "os.environ['TF_CONFIG'] = json.dumps(tf_config)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0AshGVO0_x0w"
      },
      "source": [
        "Теперь запускаем второго воркера. Запуск второго воркера запустит обучение. Так как у нас всего 2 воркеар и они оба активны, - нет необходимости в фоновом процессе:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "_ESVtyQ9_xjx"
      },
      "outputs": [],
      "source": [
        "%%bash\n",
        "python3 main.py"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hX4FA2O2AuAn"
      },
      "source": [
        "Теперь, если вы проверите логи, написанные первым воркером, вы увидите, что он участвовал в обучении модели:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "rc6hw3yTBKXX"
      },
      "outputs": [],
      "source": [
        "%%bash\n",
        "cat job_0.log"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zL79ak5PMzEg"
      },
      "source": [
        "Неудивительно, что это медленнее, чем тестовый прогон в начале этого руководства. Запуск нескольких рабочих процессов на одной машине только увеличивает накладные расходы. Целью этого примера было не сократить время, а показать обучение нескольких воркеров."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "sG5_1UgrgniF"
      },
      "outputs": [],
      "source": [
        "# Удалите TF_CONFIG и завершите все фоновые задачи, чтобы они не влияли на следующий раздел.\n",
        "\n",
        "os.environ.pop('TF_CONFIG', None)\n",
        "%killbgscripts"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9j2FJVHoUIrE"
      },
      "source": [
        "## Углубленное обучение нескольких воркеров\n",
        "\n",
        "До сих пор это мы рассматривали базовые настройки для нескольких рабочих процессов. Остальная часть этого руководства подробно рассматривает другие факторы, которые могут быть полезны для реальных случаев использования."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Rr14Vl9GR4zq"
      },
      "source": [
        "### Масштабирование данных\n",
        "\n",
        "При обучении нескольких воркеров необходимо распределение входных данных для обеспечения сходимости и производительности.\n",
        "\n",
        "В примере из предыдущего раздела используется автомасштабирование по умолчанию, предоставляемое `tf.distribute.Strategy`. Вы можете управлять масштабированием, установив `tf.data.experimental.AutoShardPolicy` для `tf.data.experimental.DistributeOptions`. Чтобы узнать больше об автоматическом масштабировании, см. [Руководство по распределенному вводу данных](https://www.tensorflow.org/tutorials/distribute/input#sharding).\n",
        "\n",
        "Вот краткий пример того, как отключить автоматическое масштабирование, чтобы каждая реплика обрабатывала каждый пример(не рекомендуется):\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "JxEtdh1vH-TF"
      },
      "outputs": [],
      "source": [
        "options = tf.data.Options()\n",
        "options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.OFF\n",
        "\n",
        "global_batch_size = 64\n",
        "multi_worker_dataset = mnist.mnist_dataset(batch_size=64)\n",
        "dataset_no_auto_shard = multi_worker_dataset.with_options(options)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "gmqvlh5LhAoU"
      },
      "source": [
        "### Оценка\n",
        "\n",
        "Если вы передадите `validation_data` в метод `model.fit`, он будет чередовать обучение и оценку для каждой эпохи. Оценка, принимающая `validation_data`, распределяется между одним и тем же набором воркеров, а результаты оценки агрегируются и доступны для всех воркеров. Подобно обучению, набор проверочных данных автоматически разделяется на уровне файла. Вам необходимо установить глобальный размер пакета для проверочного датасета и `validation_steps`. Для оценки также рекомендуется повторяемый набор данных.\n",
        "\n",
        "В качестве альтернативы вы также можете создать другую независимую задачу, которая будет периодически считывть контрольные точки и запускать оценку. Но это не рекомендуемый способ оценки, поэтому его подробности опускаются."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CsQsRfBxKMcw"
      },
      "source": [
        "### Прогноз\n",
        "Текущая реализация метода `model.predict` не работает с `MultiWorkerMirroredStrategy.`"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XVk4ftYx6JAO"
      },
      "source": [
        "### Производительность модели\n",
        "\n",
        "Теперь у вас есть модель Keras, настроеная для работы с несколькими рабочими процессами с помощью `MultiWorkerMirroredStrategy`. Для настройки производительнсоти с помощью `MultiWorkerMirroredStrategy` вы можете сделать следующее:\n",
        "\n",
        "* `MultiWorkerMirroredStrategy` предоставляет несколько [реализаций коллективного взаимодействия](https://github.com/tensorflow/tensorflow/blob/master/tensorflow/python/distribute/cross_device_ops.py). `RING` реализует группы на основе кольца, используя gRPC в качестве уровня связи между хостами. `NCCL` использует технологию [NCCL Nvidia](https://developer.nvidia.com/nccl) для реализации групп. `AUTO` откладывает выбор на время выполнения. Наилучший выбор коллективной реализации зависит от количества и типа графических процессоров, а также от сетевого соединения в кластере. Чтобы отменить `AUTO`, укажите одно из значений `RING` или `NCCL`, например `communication = tf.distribute.experimental.CollectiveCommunication.NCCL`.\n",
        "* Если возможно, преобразуйте переменные в `tf.float`. Официальная модель ResNet включает [пример](https://github.com/tensorflow/models/blob/8367cf6dabe11adf7628541706b660821f397dce/official/resnet/resnet_model.py#L466) того, как это можно сделать.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "97WhAu8uKw3j"
      },
      "source": [
        "### Отказоустойчивость\n",
        "\n",
        "При синхронном обучении весь кластер выйдет из строя, если один из рабочих процессов выйдет из строя, и на данный момент не существует механизма восстановления после сбоя. Использование Keras с `tf.distribute.Strategy` дает некоторую отказоустойчивость в случаях, когда воркеры умирают или работают нестабильно. Вы можете сделать это, сохраняя состояние обучения в распределенной файловой системе по вашему выбору. Таким образом, при перезапуске экземпляра воркера, который ранее завершился некорректно, будет восстановлено предыдущее состояние обучения.\n",
        "\n",
        "Поскольку все рабочие процессы синхронизированы с точки зрения эпох и шагов обучения, остальным воркерам придется ждать восстановления упавшего воркера, чтобы продолжить общее обучение.\n",
        "\n",
        "Примечание:\n",
        "Раньше для восстановления состояния обучения при сбое использовался обратный вызов `ModelCheckpoint`. Теперь команда TensorFlow представляет новый обратный вызов [`BackupAndRestore`](#scrollTo=kmH8uCUhfn4w). С этого момента приложения, которые полагаются на `ModelCheckpoint`, должны перейти на новый обратный вызов."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "KvHPjGlyyFt6"
      },
      "source": [
        "#### Обратный вызов ModelCheckpoint \n",
        "\n",
        "Обратный вызов `ModelCheckpoint` больше не обеспечивает поддержку отказоустойчивости, вместо него используйте обратный вызов [`BackupAndRestore`](#scrollTo=kmH8uCUhfn4w).\n",
        "\n",
        "`ModelCheckpoint` по-прежнему может использоваться для сохранения контрольных точек. Но при этом, если обучение было прервано или успешно завершилось, для продолжения обучения с контрольной точки, пользователь должен загрузить модель вручную.\n",
        "\n",
        "При желании пользователь может выбрать сохранение и восстановление модели/веса вне функции обратного вызова `ModelCheckpoint`."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "EUNV5Utc1d0s"
      },
      "source": [
        "<a id='EUNV5Utc1d0s'></a>\n",
        "### Сохранение и загрузка модели\n",
        "\n",
        "Чтобы сохранить модель с помощью `model.save` или` tf.saved_model.save`, место сохранения должно быть разным для каждого воркера. На машинах обычных воркеров вам нужно будет сохранить модель во временной директории, а на машине с главным воркером - в предоставленный каталог модели. Временные каталоги на воркере должны быть уникальными, чтобы избежать того, что несколько воркеров пытаются писать в одно и то же место. Модели, сохраненные во всех каталогах, идентичны, и обычно для восстановления или обслуживания следует ссылаться только на модель, сохраненную главным воркером. У вас должна быть некоторая логика очистки, которая удаляет временные каталоги, созданные воркерами, после завершения вашего обучения.\n",
        "\n",
        "Причина, по которой вам нужно сохранять модель на главном и обычных воркерах одновременно, заключается в том, что вы можете агрегировать переменные во время записи контрольной точки, и для этого требуется, чтобы и руководитель, и воркеры участвовали в протоколе объединения расчетов. С другой стороны, разрешение главному и остальным воркерам сохранять модели в один и тот же каталог приведет к ошибкам записи.\n",
        "\n",
        "С `MultiWorkerMirroredStrategy` программа запускается для каждого рабочего, и чтобы узнать, является ли текущий работник главным, она использует объект определителя кластера, который имеет атрибуты `task_type` и `task_id`. `task_type` сообщает вам текущее задание (например, `worker_1`), а `task_id` сообщает идентификатор работника, например `1`. Воркер с идентификатором `0` назначается главным.\n",
        "\n",
        "В приведенном ниже фрагменте кода `write_filepath` возвращает путь к файлу для записи, который зависит от идентификатора воркера. В случае руководителя(работника с идентификатором `0`) он записывает в исходный путь к файлу; для других он создает временный каталог с идентификатором в пути к каталогу:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "XQfGkmg-pfCY"
      },
      "outputs": [],
      "source": [
        "model_path = '/tmp/keras-model'\n",
        "\n",
        "def _is_chief(task_type, task_id):\n",
        "  # If `task_type` is None, this may be operating as single worker, which works\n",
        "  # effectively as chief.\n",
        "  return task_type is None or task_type == 'chief' or (\n",
        "            task_type == 'worker' and task_id == 0)\n",
        "\n",
        "def _get_temp_dir(dirpath, task_id):\n",
        "  base_dirpath = 'workertemp_' + str(task_id)\n",
        "  temp_dir = os.path.join(dirpath, base_dirpath)\n",
        "  tf.io.gfile.makedirs(temp_dir)\n",
        "  return temp_dir\n",
        "\n",
        "def write_filepath(filepath, task_type, task_id):\n",
        "  dirpath = os.path.dirname(filepath)\n",
        "  base = os.path.basename(filepath)\n",
        "  if not _is_chief(task_type, task_id):\n",
        "    dirpath = _get_temp_dir(dirpath, task_id)\n",
        "  return os.path.join(dirpath, base)\n",
        "\n",
        "task_type, task_id = (strategy.cluster_resolver.task_type,\n",
        "                      strategy.cluster_resolver.task_id)\n",
        "write_model_path = write_filepath(model_path, task_type, task_id)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hs0_agYR_qKm"
      },
      "source": [
        "Теперь вы готовы сохранить:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "J-yA3BYG_vTs"
      },
      "outputs": [],
      "source": [
        "multi_worker_model.save(write_model_path)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "8LXUVVl9_v5x"
      },
      "source": [
        "Как описано выше, модель загружается только из пути, в который сохранял главный воркер, поэтому давайте удалим временные директории, которые сохранили остальные воркеры:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "aJTyu-97ABpY"
      },
      "outputs": [],
      "source": [
        "if not _is_chief(task_type, task_id):\n",
        "  tf.io.gfile.rmtree(os.path.dirname(write_model_path))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Nr-2PKlHAPBT"
      },
      "source": [
        "Теперь, когда нужно загрузить модель, воспользуемся API `tf.keras.models.load_model` и продолжим обучение. Предположим, что для загрузки и продолжения обучения используется только один воркер, в таком случае вам не нужно вызывать `tf.keras.models.load_model` внутри `strategy.scope()`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "iUZna-JKAOrX"
      },
      "outputs": [],
      "source": [
        "loaded_model = tf.keras.models.load_model(model_path)\n",
        "\n",
        "# Модель восстановлена и мы можем продолжить обучение.\n",
        "loaded_model.fit(single_worker_dataset, epochs=2, steps_per_epoch=20)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YJ1fmxmTpocS"
      },
      "source": [
        "### Сохранение и восстановление контрольной точки(чекпойнта)\n",
        "\n",
        "Контрольные точки позволяют сохранять веса модели и восстанавливать их без необходимости сохранять и восстанавливать всю модель. Далее вы создадите один `tf.train.Checkpoint` для отслеживания модели, которая управляется `tf.train.CheckpointManager` и сохраняет только последний чекпойнт."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "_1-RYaB5xnNH"
      },
      "outputs": [],
      "source": [
        "checkpoint_dir = '/tmp/ckpt'\n",
        "\n",
        "checkpoint = tf.train.Checkpoint(model=multi_worker_model)\n",
        "write_checkpoint_dir = write_filepath(checkpoint_dir, task_type, task_id)\n",
        "checkpoint_manager = tf.train.CheckpointManager(\n",
        "    checkpoint, directory=write_checkpoint_dir, max_to_keep=1)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7oBpPCRsW1MF"
      },
      "source": [
        "После создания `CheckpointManager` вы сохраняете чекпойнты, а затем удалите все чекпойнты, созданные обычными воркерами(не главным)."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "l1ZXG_GbWzLp"
      },
      "outputs": [],
      "source": [
        "checkpoint_manager.save()\n",
        "if not _is_chief(task_type, task_id):\n",
        "  tf.io.gfile.rmtree(write_checkpoint_dir)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "RO7cbN40XD5v"
      },
      "source": [
        "Теперь, когда вам нужно восстановить чекпойнт, вы можете найти последний с помощью удобной функции `tf.train.latest_checkpoint`. После восстановления чекпойнта можно продолжить обучение."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "NJW7vtknXFEH"
      },
      "outputs": [],
      "source": [
        "latest_checkpoint = tf.train.latest_checkpoint(checkpoint_dir)\n",
        "checkpoint.restore(latest_checkpoint)\n",
        "multi_worker_model.fit(multi_worker_dataset, epochs=2, steps_per_epoch=20)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kmH8uCUhfn4w"
      },
      "source": [
        "#### Обратный вызов BackupAndRestore\n",
        "\n",
        "Примечание: обратный вызов `tf.keras.callbacks.experimental.BackupAndRestore` доступен только в ночных сборках tf-nightly.\n",
        "\n",
        "Обратный вызов `BackupAndRestore` обеспечивает функциональность отказоустойчивости, создавая резервную копию модели и текущего номера эпохи во временном файле, переданном в аргументе `backup_dir` для `BackupAndRestore`. Это делается в конце каждой эпохи.\n",
        "\n",
        "Если задания прерываются и перезапускаются, `BackupAndRestore` восстанавливает последнюю контрольную точку, и обучение продолжается с начала прерванной эпохи. Любое частичное обучение, уже выполненное в незавершенную эпоху до прерывания, будет отброшено, и не повлияет на окончательное состояние модели.\n",
        "\n",
        "Чтобы `BackupAndRestore`, передайте его экземпляр при вызове `tf.keras.Model.fit()`.\n",
        "\n",
        "При использовании `MultiWorkerMirroredStrategy`, если воркер прерывается, весь кластер приостанавливается, пока прерванный воркер не будет перезапущен. Другие воркеры также перезапустятся, и прерванный воркер снова присоединится к кластеру. Затем каждый воркер читает файл контрольной точки, который был ранее сохранен, и загружает его прежнее состояние, тем самым позволяя кластеру синхронизироваться. Затем обучение продолжается.\n",
        "\n",
        "Обратный вызов `BackupAndRestore` для сохранения и восстановления состояния обучения использует `CheckpointManager`, который, в свою очередь генерирует файл, называемый контрольной точкой, и отслеживает существующие контрольные точки вместе с последней. По этой причине не следует повторно использовать одно и тоже значение `backup_dir`, чтобы избежать конфликта имен.\n",
        "\n",
        "В настоящее время обратный вызов `BackupAndRestore` поддерживает одного воркера без стратегии, одного воркера с `MirroredStrategy`, и нескольких воркеров с помощью `MultiWorkerMirroredStrategy`.\n",
        "Ниже приведены два примера обучения нескольких воркеров и обучение одного воркера."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CYdzZi4Qs1jz"
      },
      "outputs": [],
      "source": [
        "# Multi-worker training with MultiWorkerMirroredStrategy.\n",
        "\n",
        "callbacks = [tf.keras.callbacks.experimental.BackupAndRestore(backup_dir='/tmp/backup')]\n",
        "with strategy.scope():\n",
        "  multi_worker_model = mnist.build_and_compile_cnn_model()\n",
        "multi_worker_model.fit(multi_worker_dataset,\n",
        "                       epochs=3,\n",
        "                       steps_per_epoch=70,\n",
        "                       callbacks=callbacks)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rIV5_3ebzXmB"
      },
      "source": [
        "Если вы проверите каталог `backup_dir`, который вы указали в `BackupAndRestore`, вы увидите временно сгенерированные файлы контрольных точек. Эти файлы нужны для восстановления ранее утерянных экземпляров, и они будут удалены библиотекой в конце `tf.keras.Model.fit()`, после успешного завершения вашего обучения.\n",
        "\n",
        "Примечание. В настоящее время `BackupAndRestore` поддерживает только активный режим. В графическом режиме рассмотрите возможность использования [Сохранение и загрузка модели](#scrollTo=EUNV5Utc1d0s), упомянутого выше."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ega2hdOQEmy_"
      },
      "source": [
        "## Смотрите также\n",
        "1. В руководстве [Распределенное обучение в TensorFlow](https://www.tensorflow.org/guide/distributed_training) представлен обзор доступных стратегий распределения.\n",
        "2. [Официальные модели](https://github.com/tensorflow/models/tree/master/official), многие из которых можно настроить для запуска с несколькими стратегиями.\n",
        "3. Раздел [Производительность](../../guide/function.ipynb) предоставляет информацию о других стратегиях и [инструментах](../../guide/profiler.md), которые можно использовать для оптимизации производительности ваших моделей TensorFlow.\n"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "multi_worker_with_keras.ipynb",
      "toc_visible": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}
